<!DOCTYPE html>
<html lang="en-US">
  <head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width,initial-scale=1">
    <title>kafka 生产者 | Oliver知识收集站</title>
    <meta name="generator" content="VuePress 1.9.7">
    
    <meta name="description" content="享受着互联网广泛知识，并加以记录，日积月累让它成为一个档案处！">
    <meta name="viewport" content="width=device-width,initial-scale=1,user-scalable=no">
    
    <link rel="preload" href="/oliver-vuepress/assets/css/0.styles.4ea20d86.css" as="style"><link rel="preload" href="/oliver-vuepress/assets/js/app.c21e6ffc.js" as="script"><link rel="preload" href="/oliver-vuepress/assets/js/3.6dd9a2a1.js" as="script"><link rel="preload" href="/oliver-vuepress/assets/js/1.898920d0.js" as="script"><link rel="preload" href="/oliver-vuepress/assets/js/44.10d7fe47.js" as="script"><link rel="prefetch" href="/oliver-vuepress/assets/js/10.41b2bf91.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/11.a95c117d.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/12.8607f0e1.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/13.a52d6846.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/14.249b4e52.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/15.d458d12e.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/16.ba334206.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/17.1b91c9fa.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/18.e2ea2eb5.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/19.bf0e2553.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/20.268bd174.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/21.cd1bbed5.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/22.da4bc7f7.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/23.12f0c72f.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/24.b7886742.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/25.6e71af85.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/26.5c127243.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/27.e98fd8bf.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/28.ce83b09c.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/29.50398f0f.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/30.05e1339c.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/31.ef4b13fb.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/32.ba5f8351.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/33.3902db0a.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/34.36a05884.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/35.87215872.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/36.db360c58.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/37.402e5374.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/38.c9228dd8.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/39.72ba5d1f.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/4.7bb03d47.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/40.7e7949bf.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/41.c0d5b947.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/42.d9984467.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/43.e6a43668.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/45.f692ec2d.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/46.9b920343.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/47.8e3d94f9.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/48.7d356e5b.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/49.b0df6271.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/5.1fa544da.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/50.805e1466.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/51.1b31d40e.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/52.44e69a41.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/53.da1def53.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/54.6569f7db.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/55.5fc3de47.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/56.da649377.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/57.6ff15ed4.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/58.a62f6424.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/59.f68ae517.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/6.f5bd8e9b.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/60.dda416bc.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/61.4e0c719f.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/62.8c5ef01e.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/63.7089eb8b.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/64.b5ec150d.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/65.6720cda4.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/66.4ee90e29.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/67.cc4b0c6d.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/7.d5950c53.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/8.382fb3a5.js"><link rel="prefetch" href="/oliver-vuepress/assets/js/9.d593f4c1.js">
    <link rel="stylesheet" href="/oliver-vuepress/assets/css/0.styles.4ea20d86.css">
  </head>
  <body>
    <div id="app" data-server-rendered="true"><div class="theme-container" data-v-130b300a><div data-v-130b300a><div class="password-shadow password-wrapper-out" style="display:none;" data-v-25ba6db2 data-v-130b300a data-v-130b300a><h3 class="title" data-v-25ba6db2 data-v-25ba6db2>Oliver知识收集站</h3> <p class="description" data-v-25ba6db2 data-v-25ba6db2>享受着互联网广泛知识，并加以记录，日积月累让它成为一个档案处！</p> <label id="box" class="inputBox" data-v-25ba6db2 data-v-25ba6db2><input type="password" value="" data-v-25ba6db2> <span data-v-25ba6db2>Konck! Knock!</span> <button data-v-25ba6db2>OK</button></label> <div class="footer" data-v-25ba6db2 data-v-25ba6db2><span data-v-25ba6db2><i class="iconfont reco-theme" data-v-25ba6db2></i> <a target="blank" href="https://vuepress-theme-reco.recoluan.com" data-v-25ba6db2>vuePress-theme-reco</a></span> <span data-v-25ba6db2><i class="iconfont reco-copyright" data-v-25ba6db2></i> <a data-v-25ba6db2><span data-v-25ba6db2>oliver.shi</span>
            
          <!---->
          2022
        </a></span></div></div> <div class="hide" data-v-130b300a><header class="navbar" data-v-130b300a><div class="sidebar-button"><svg xmlns="http://www.w3.org/2000/svg" aria-hidden="true" role="img" viewBox="0 0 448 512" class="icon"><path fill="currentColor" d="M436 124H12c-6.627 0-12-5.373-12-12V80c0-6.627 5.373-12 12-12h424c6.627 0 12 5.373 12 12v32c0 6.627-5.373 12-12 12zm0 160H12c-6.627 0-12-5.373-12-12v-32c0-6.627 5.373-12 12-12h424c6.627 0 12 5.373 12 12v32c0 6.627-5.373 12-12 12zm0 160H12c-6.627 0-12-5.373-12-12v-32c0-6.627 5.373-12 12-12h424c6.627 0 12 5.373 12 12v32c0 6.627-5.373 12-12 12z"></path></svg></div> <a href="/oliver-vuepress/" class="home-link router-link-active"><!----> <span class="site-name">Oliver知识收集站</span></a> <div class="links"><div class="color-picker"><a class="color-button"><i class="iconfont reco-color"></i></a> <div class="color-picker-menu" style="display:none;"><div class="mode-options"><h4 class="title">Choose mode</h4> <ul class="color-mode-options"><li class="dark">dark</li><li class="auto active">auto</li><li class="light">light</li></ul></div></div></div> <div class="search-box"><i class="iconfont reco-search"></i> <input aria-label="Search" autocomplete="off" spellcheck="false" value=""> <!----></div> <nav class="nav-links can-hide"><div class="nav-item"><a href="/oliver-vuepress/" class="nav-link"><i class="undefined"></i>
  主页
</a></div><div class="nav-item"><div class="dropdown-wrapper"><a class="dropdown-title"><span class="title"><i class="undefined"></i>
      Java
    </span> <span class="arrow right"></span></a> <ul class="nav-dropdown" style="display:none;"><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/java/basics/" class="nav-link"><i class="undefined"></i>
  基础
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/java/concurrent/" class="nav-link"><i class="undefined"></i>
  并发
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/java/jvm/jvm.html" class="nav-link"><i class="undefined"></i>
  JVM
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/java/other/" class="nav-link"><i class="undefined"></i>
  杂
</a></li></ul></div></div><div class="nav-item"><a href="/oliver-vuepress/articles/spring/first.html" class="nav-link"><i class="undefined"></i>
  Spring
</a></div><div class="nav-item"><div class="dropdown-wrapper"><a class="dropdown-title"><span class="title"><i class="undefined"></i>
      中间件
    </span> <span class="arrow right"></span></a> <ul class="nav-dropdown" style="display:none;"><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/middleware/redis/redis.html" class="nav-link"><i class="undefined"></i>
  Redis
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/middleware/kafka/framework.html" class="nav-link"><i class="undefined"></i>
  Kafka
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/middleware/zookeeper.html" class="nav-link"><i class="undefined"></i>
  Zookeeper
</a></li></ul></div></div><div class="nav-item"><a href="/oliver-vuepress/articles/algorithm/" class="nav-link"><i class="undefined"></i>
  算法
</a></div><div class="nav-item"><a href="/oliver-vuepress/timeline/" class="nav-link"><i class="iconfont reco-date"></i>
  TimeLine
</a></div><div class="nav-item"><div class="dropdown-wrapper"><a class="dropdown-title"><span class="title"><i class="undefined"></i>
      收集站
    </span> <span class="arrow right"></span></a> <ul class="nav-dropdown" style="display:none;"><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/article/first.html" class="nav-link"><i class="undefined"></i>
  技术好文
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/book/first.html" class="nav-link"><i class="undefined"></i>
  书籍
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/assembly/first.html" class="nav-link"><i class="undefined"></i>
  优秀开发组件
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/software/first.html" class="nav-link"><i class="undefined"></i>
  软件
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/plugin/first.html" class="nav-link"><i class="undefined"></i>
  插件
</a></li></ul></div></div> <!----></nav></div></header> <div class="sidebar-mask" data-v-130b300a></div> <aside class="sidebar" data-v-130b300a><div class="personal-info-wrapper" data-v-39576ba9 data-v-130b300a><!----> <h3 class="name" data-v-39576ba9>
    oliver.shi
  </h3> <div class="num" data-v-39576ba9><div data-v-39576ba9><h3 data-v-39576ba9>52</h3> <h6 data-v-39576ba9>Articles</h6></div> <div data-v-39576ba9><h3 data-v-39576ba9>6</h3> <h6 data-v-39576ba9>Tags</h6></div></div> <ul class="social-links" data-v-39576ba9></ul> <hr data-v-39576ba9></div> <nav class="nav-links"><div class="nav-item"><a href="/oliver-vuepress/" class="nav-link"><i class="undefined"></i>
  主页
</a></div><div class="nav-item"><div class="dropdown-wrapper"><a class="dropdown-title"><span class="title"><i class="undefined"></i>
      Java
    </span> <span class="arrow right"></span></a> <ul class="nav-dropdown" style="display:none;"><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/java/basics/" class="nav-link"><i class="undefined"></i>
  基础
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/java/concurrent/" class="nav-link"><i class="undefined"></i>
  并发
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/java/jvm/jvm.html" class="nav-link"><i class="undefined"></i>
  JVM
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/java/other/" class="nav-link"><i class="undefined"></i>
  杂
</a></li></ul></div></div><div class="nav-item"><a href="/oliver-vuepress/articles/spring/first.html" class="nav-link"><i class="undefined"></i>
  Spring
</a></div><div class="nav-item"><div class="dropdown-wrapper"><a class="dropdown-title"><span class="title"><i class="undefined"></i>
      中间件
    </span> <span class="arrow right"></span></a> <ul class="nav-dropdown" style="display:none;"><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/middleware/redis/redis.html" class="nav-link"><i class="undefined"></i>
  Redis
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/middleware/kafka/framework.html" class="nav-link"><i class="undefined"></i>
  Kafka
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/middleware/zookeeper.html" class="nav-link"><i class="undefined"></i>
  Zookeeper
</a></li></ul></div></div><div class="nav-item"><a href="/oliver-vuepress/articles/algorithm/" class="nav-link"><i class="undefined"></i>
  算法
</a></div><div class="nav-item"><a href="/oliver-vuepress/timeline/" class="nav-link"><i class="iconfont reco-date"></i>
  TimeLine
</a></div><div class="nav-item"><div class="dropdown-wrapper"><a class="dropdown-title"><span class="title"><i class="undefined"></i>
      收集站
    </span> <span class="arrow right"></span></a> <ul class="nav-dropdown" style="display:none;"><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/article/first.html" class="nav-link"><i class="undefined"></i>
  技术好文
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/book/first.html" class="nav-link"><i class="undefined"></i>
  书籍
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/assembly/first.html" class="nav-link"><i class="undefined"></i>
  优秀开发组件
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/software/first.html" class="nav-link"><i class="undefined"></i>
  软件
</a></li><li class="dropdown-item"><!----> <a href="/oliver-vuepress/articles/collect/plugin/first.html" class="nav-link"><i class="undefined"></i>
  插件
</a></li></ul></div></div> <!----></nav> <ul class="sidebar-links"><li><section class="sidebar-group collapsable depth-0"><p class="sidebar-heading open"><span>Kafka</span> <span class="arrow down"></span></p> <ul class="sidebar-links sidebar-group-items"><li><a href="/oliver-vuepress/articles/middleware/kafka/framework.html" class="sidebar-link">kafka 架构体系</a></li><li><a href="/oliver-vuepress/articles/middleware/kafka/producer.html" aria-current="page" class="active sidebar-link">kafka 生产者</a></li><li><a href="/oliver-vuepress/articles/middleware/kafka/consumer.html" class="sidebar-link">kafka 消费者</a></li><li><a href="/oliver-vuepress/articles/middleware/kafka/log.html" class="sidebar-link">kafka 日志</a></li><li><a href="/oliver-vuepress/articles/middleware/kafka/why-fast.html" class="sidebar-link">kafka 为什么快</a></li><li><a href="/oliver-vuepress/articles/middleware/kafka/application.html" class="sidebar-link">kafka的优缺点&amp; 应用场景</a></li></ul></section></li></ul> </aside> <div class="password-shadow password-wrapper-in" style="display:none;" data-v-25ba6db2 data-v-130b300a><h3 class="title" data-v-25ba6db2 data-v-25ba6db2>kafka 生产者</h3> <!----> <label id="box" class="inputBox" data-v-25ba6db2 data-v-25ba6db2><input type="password" value="" data-v-25ba6db2> <span data-v-25ba6db2>Konck! Knock!</span> <button data-v-25ba6db2>OK</button></label> <div class="footer" data-v-25ba6db2 data-v-25ba6db2><span data-v-25ba6db2><i class="iconfont reco-theme" data-v-25ba6db2></i> <a target="blank" href="https://vuepress-theme-reco.recoluan.com" data-v-25ba6db2>vuePress-theme-reco</a></span> <span data-v-25ba6db2><i class="iconfont reco-copyright" data-v-25ba6db2></i> <a data-v-25ba6db2><span data-v-25ba6db2>oliver.shi</span>
            
          <!---->
          2022
        </a></span></div></div> <div data-v-130b300a><main class="page"><section><div class="page-title"><h1 class="title">kafka 生产者</h1> <div data-v-f875f3fc><i class="iconfont reco-account" data-v-f875f3fc><span data-v-f875f3fc>oliver.shi</span></i> <i class="iconfont reco-date" data-v-f875f3fc><span data-v-f875f3fc>12/5/2021</span></i> <!----> <i class="tags iconfont reco-tag" data-v-f875f3fc><span class="tag-item" data-v-f875f3fc>消息队列</span><span class="tag-item" data-v-f875f3fc>kafka</span></i></div></div> <div class="theme-reco-content content__default"><blockquote><p>生产者： 负责将消息发送到kafka中，生产者核心通过 主线程 和 <code>Sender</code>  线程 将发送拆分成两部<br>
并使用的批处理 和 压缩 技术 大大的提高了 消息发送的效率</p></blockquote> <h2 id="kafka-批量发送机制"><a href="#kafka-批量发送机制" class="header-anchor">#</a> kafka 批量发送机制</h2> <blockquote><p>Kafak 发送的消息主要会将数据打包按批次发送，主要靠两个参数决定，那个条件先满足，就算那个</p></blockquote> <ul><li><code>batch.size</code> :  每批数据大小 默认是16KB</li> <li><code>linger.ms</code>：设置延时时间 默认是 0 （即有消息则直接发送）</li></ul> <p><strong>注：两个参数同时设置时，当 <code>batch.size</code>  数据没有填满时，等待 <code>linger.ms</code>  设置的时间后，进行发送数据</strong></p> <p><code>linger.ms</code>  会增加 消息发送的延时性，负载低时 消息会延时发送
负载高时，batch被填满的时间远低于 <code>linger.ms</code>, 此时参数起不到作用</p> <p><strong>生产者架构图</strong></p> <img src="/oliver-vuepress/middleware/kafka/producer1.jpg" alt="foo"> <p><a href="https://www.processon.com/view/6197b4db1efad406f8795981" target="_blank" rel="noopener noreferrer">详细的流程图解析<span><svg xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15" class="icon outbound"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path> <polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg> <span class="sr-only">(opens new window)</span></span></a></p> <p><strong>简述发送消息流程</strong></p> <ul><li>主线程：
<ul><li>消息经过 拦截器 -&gt; 序列化器 -&gt; 分区器 写到 消息累加器中 指定分区队列的尾部 （RecoderAccumulator为每个分区都维护了一个 <code>Deque&lt;ProducerBatch&gt;</code> 类型的双端队列）</li></ul></li> <li>Sender线程
<ul><li>异步监听获取 消息累计器中的数据</li> <li>将 数据 转换成 Broker （node 即 对应的服务器地址）对应的 request  Map集合发送结构体 存储在 <code>NetworkClient</code> 中 等待发送</li> <li>成功发送后，接收到发送成功的回调消息后，清理 <code>NetworkClient</code> 和 <code>消息累加器</code>  中存在的数据</li></ul></li></ul> <h2 id="生产者集群架构"><a href="#生产者集群架构" class="header-anchor">#</a> 生产者集群架构</h2> <h3 id="拦截器"><a href="#拦截器" class="header-anchor">#</a> 拦截器</h3> <blockquote><p>主要作用 实现 Client端的定制化控制逻辑，一般拦截器会对 消息发送前 以及 Producer 回调逻辑进行拦截</p></blockquote> <ul><li>一个生产者 可以指定多个拦截器，多个拦截的线程安全 需要自己的应用程序保证</li></ul> <ul><li>拦截器抛出异常后，不会向上传递，可能会直接爆出异常</li></ul> <p><strong>源码</strong></p> <div class="language-Java extra-class"><pre class="language-java"><code><span class="token keyword">public</span> <span class="token keyword">interface</span> <span class="token class-name">ProducerInterceptor</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> <span class="token keyword">extends</span> <span class="token class-name">Configurable</span> <span class="token punctuation">{</span>
  	<span class="token comment">// 在消息序列化 和 计算分区之前，可以对消息进行定制化操作</span>
    <span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> <span class="token function">onSend</span><span class="token punctuation">(</span><span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> var1<span class="token punctuation">)</span><span class="token punctuation">;</span>
	
  	<span class="token comment">// 会在消息被应答（Acknowledgement）之前或消息发送）之前或消息发送失败时调用生产者拦截器的onAcknowledgement（）方法，优先于用户设定的Callback之前执行这个方法运行在 Producer I/O 线程中，所以这个方法中实现的代码逻辑越简单越好则会影响消息的发送速度</span>
    <span class="token keyword">void</span> <span class="token function">onAcknowledgement</span><span class="token punctuation">(</span><span class="token class-name">RecordMetadata</span> var1<span class="token punctuation">,</span> <span class="token class-name">Exception</span> var2<span class="token punctuation">)</span><span class="token punctuation">;</span>
		
    <span class="token comment">// 用于在关闭拦截器时执行一些资源的清理工作</span>
    <span class="token keyword">void</span> <span class="token function">close</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre></div><p><strong>注</strong> 可通过实现  <code>org.apache.kafka.clients.producer.ProducerInterceptor</code>  接口来实现 自定义拦截器</p> <h3 id="序列化器"><a href="#序列化器" class="header-anchor">#</a> 序列化器</h3> <blockquote><p>kafka 中 数据都是 字节数组， 所以需要在消息发送前，将 数据 序列化为 字节数组</p></blockquote> <p><strong>基本的序列化器</strong></p> <ul><li>ByteArraySerializer</li> <li>ByteBufferSerializer</li> <li>ByteSerializer</li> <li>DoubleSerializer</li> <li>FloatSerializer</li> <li>IntegerSerializer</li> <li>LongSerializer</li> <li>ShortSerializer</li></ul> <p><strong>注</strong> 可通过实现  <code>org.apache.kafka.common.serialization.Serializer</code>  接口来实现 自定义序列化器</p> <h3 id="分区器"><a href="#分区器" class="header-anchor">#</a> 分区器</h3> <blockquote><p>用于确认数据应该被发送到 <code>topic</code>  中的那个分区</p></blockquote> <p>kafka 默认分区器 <code>DefaultPartitioner</code>  中提供了三种策略:</p> <ol><li>通过 指定 分区号，发往指定的分区</li> <li>通过 指定 key, 对 key 进行 hash 取模 得到 分区号，发往指定的分区</li> <li>通过轮询方式，进行分配分区 （可以实现绝对的分配均衡）</li></ol> <p><strong>注</strong>  可以通过实现 <code>org.apache.kafka.clients.producer.Partitioner</code>   进行自定分区</p> <h3 id="消息累加器"><a href="#消息累加器" class="header-anchor">#</a> 消息累加器</h3> <blockquote><p>kafka发送消息时，并不是直接将消息从客户端通过网络发送给服务器端，而是先将消息存储在客户端的记录收集器中，当队列满了 或者 发送时间已到的时候才会去发送，这个记录收集器就是 <code>RecordAccumulator</code></p></blockquote> <img src="/oliver-vuepress/middleware/kafka/producer2.jpg" alt="foo"> <p><strong>核心参数</strong> <strong>buffer.memory</strong></p> <ul><li>默认值：32M</li> <li>该参数决定了 消息累加器的 整个缓存区的大小，当内存满时，消费发送 <code>Send()</code>  方法会被堵塞</li></ul> <p><strong>max.block.ms</strong></p> <ul><li>默认值：60000, 即 60秒</li> <li>该参数决定了 当内存满时，消息发送被堵塞的时间长度</li> <li>当超过堵塞的时间长度，则会抛出异常</li></ul> <p><strong>batch.size</strong></p> <ul><li>默认值：16KB</li> <li>该参数决定了 每批次数据达到16K 发送一次 ， 而kafka 将每一批次打包成一个 <code>ProducerBatch</code> 对象</li> <li>每次发送消息，就是在像 <code>ProducerBatch</code>   中插入数据</li></ul> <p>::插入ProducerBatch的逻辑::</p> <img src="/oliver-vuepress/middleware/kafka/producer3.jpg" alt="foo"> <p><strong>数据结构</strong></p> <div class="language-Java extra-class"><pre class="language-java"><code>
<span class="token comment">// kafka 生产者</span>
<span class="token keyword">public</span> <span class="token keyword">class</span> <span class="token class-name">KafkaProducer</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> <span class="token keyword">implements</span> <span class="token class-name">Producer</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> <span class="token punctuation">{</span>
	  <span class="token comment">// 消息缓存区</span>
    <span class="token keyword">private</span> <span class="token keyword">final</span> <span class="token class-name">RecordAccumulator</span> accumulator<span class="token punctuation">;</span>
<span class="token punctuation">}</span>

<span class="token comment">// 消息缓存区</span>
<span class="token keyword">public</span> <span class="token keyword">final</span> <span class="token keyword">class</span> <span class="token class-name">RecordAccumulator</span> <span class="token punctuation">{</span>
	  
  <span class="token comment">/**
   * 具体存放数据的Map集合
   * key: TopicPartition   主题信息对象 【主题、分区】
   * value: Deque&lt;ProducerBatch&gt;  一个 由 ProducerBatch 组成的 消息双端队列
	 * 一个 producerBatch 一组批量的数据 默认为 16K
   */</span>
    <span class="token keyword">private</span> <span class="token keyword">final</span> <span class="token class-name">ConcurrentMap</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">TopicPartition</span><span class="token punctuation">,</span> <span class="token class-name">Deque</span><span class="token punctuation">&lt;</span><span class="token class-name">ProducerBatch</span><span class="token punctuation">&gt;</span><span class="token punctuation">&gt;</span></span> batches<span class="token punctuation">;</span>
<span class="token punctuation">}</span>

<span class="token comment">// 消息结构体</span>
<span class="token keyword">public</span> <span class="token keyword">class</span> <span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> <span class="token punctuation">{</span>
    <span class="token keyword">private</span> <span class="token keyword">final</span> <span class="token class-name">String</span> topic<span class="token punctuation">;</span>  			<span class="token comment">// 主题</span>
    <span class="token keyword">private</span> <span class="token keyword">final</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">;</span>  		<span class="token comment">// 分区号	</span>
    <span class="token keyword">private</span> <span class="token keyword">final</span> <span class="token class-name">Headers</span> headers<span class="token punctuation">;</span>			<span class="token comment">// 消息头部</span>
    <span class="token keyword">private</span> <span class="token keyword">final</span> <span class="token class-name">K</span> key<span class="token punctuation">;</span>						<span class="token comment">// 键	</span>
    <span class="token keyword">private</span> <span class="token keyword">final</span> <span class="token class-name">V</span> value<span class="token punctuation">;</span> 					<span class="token comment">// 值</span>
    <span class="token keyword">private</span> <span class="token keyword">final</span> <span class="token class-name">Long</span> timestamp<span class="token punctuation">;</span>     		<span class="token comment">// 消息的时间戳</span>
<span class="token punctuation">}</span>
</code></pre></div><h2 id="ack策略"><a href="#ack策略" class="header-anchor">#</a> ack策略</h2> <blockquote><p>Kafka producer 有三种 ack 机制，为解决消息的重复和丢失。可以在初始化 producer时在 config 中进行配置，默认值：acks=1</p></blockquote> <p><strong>具体参数类型</strong></p> <ul><li>acks = 0
<ul><li>生产者不等待broker的任何消息确认。只要将消息放到了socket的缓冲区，就认为消息已发送</li> <li>retries【配置重试参数】 设置也不起作用，因为客户端不关心消息是否发送失败。</li></ul></li> <li>acks = 1
<ul><li>leader分区 将记录写到它本地日志，就响应客户端确认消息，而不等待follower副本的确认。</li> <li>如果leader确认了消息就宕机，则可能会丢失消息，因为follower副本可能还没来得及同步该消息。</li></ul></li> <li>acks = all (-1)
<ul><li>leader等待所有同步的副本确认该消息。保证了只要有一个同步副本存在，消息就不会丢失。这是最强的可用性保证</li></ul></li></ul> <p>::<strong>最小同步副本</strong>::</p> <p><a href="https://zhuanlan.zhihu.com/p/165988882" target="_blank" rel="noopener noreferrer">Kafka生产者ack机制剖析 - 知乎<span><svg xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15" class="icon outbound"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path> <polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg> <span class="sr-only">(opens new window)</span></span></a></p> <h2 id="重试机制"><a href="#重试机制" class="header-anchor">#</a> 重试机制</h2> <blockquote><p>Kafka 提供了生成消息时，可以设置对失败消息进行重试的机制</p></blockquote> <p><strong>核心参数</strong></p> <ul><li>retries：重试次数</li> <li>retry.backoff.ms：间隔多少时间重试一次</li></ul> <p><strong>注</strong>  Kafka 中 重试机制 会破坏原有的顺序消费，如果需要保证顺序消费。如果原本需要顺序发送的信息，失败后会变成异步定时发送，特殊情况会影响原有的前后顺序</p> <p><strong>解决方式</strong>
需要进行设置 <code>max.in.flight.requests.per.connection</code> 参数。 该参数 是指定服务器每次可以发送消息的批次数，当设置为 1 时，则每次只能有一个批次进行发送，当出现发送失败重试时，则会堵塞请求队列</p></div></section> <footer class="page-edit"><!----> <div class="last-updated"><span class="prefix">上次更新: </span> <span class="time">4/21/2022, 8:17:59 AM</span></div></footer> <div class="page-nav"><p class="inner"><span class="prev"><a href="/oliver-vuepress/articles/middleware/kafka/framework.html" class="prev">
            kafka 架构体系
          </a></span> <span class="next"><a href="/oliver-vuepress/articles/middleware/kafka/consumer.html">
            kafka 消费者
          </a></span></p></div> <div class="comments-wrapper"><!----></div> <ul class="side-bar sub-sidebar-wrapper" style="width:12rem;" data-v-cb1513f6><li class="level-2" data-v-cb1513f6><a href="/oliver-vuepress/articles/middleware/kafka/producer.html#kafka-批量发送机制" class="sidebar-link reco-side-kafka-批量发送机制" data-v-cb1513f6>kafka 批量发送机制</a></li><li class="level-2" data-v-cb1513f6><a href="/oliver-vuepress/articles/middleware/kafka/producer.html#生产者集群架构" class="sidebar-link reco-side-生产者集群架构" data-v-cb1513f6>生产者集群架构</a></li><li class="level-3" data-v-cb1513f6><a href="/oliver-vuepress/articles/middleware/kafka/producer.html#拦截器" class="sidebar-link reco-side-拦截器" data-v-cb1513f6>拦截器</a></li><li class="level-3" data-v-cb1513f6><a href="/oliver-vuepress/articles/middleware/kafka/producer.html#序列化器" class="sidebar-link reco-side-序列化器" data-v-cb1513f6>序列化器</a></li><li class="level-3" data-v-cb1513f6><a href="/oliver-vuepress/articles/middleware/kafka/producer.html#分区器" class="sidebar-link reco-side-分区器" data-v-cb1513f6>分区器</a></li><li class="level-3" data-v-cb1513f6><a href="/oliver-vuepress/articles/middleware/kafka/producer.html#消息累加器" class="sidebar-link reco-side-消息累加器" data-v-cb1513f6>消息累加器</a></li><li class="level-2" data-v-cb1513f6><a href="/oliver-vuepress/articles/middleware/kafka/producer.html#ack策略" class="sidebar-link reco-side-ack策略" data-v-cb1513f6>ack策略</a></li><li class="level-2" data-v-cb1513f6><a href="/oliver-vuepress/articles/middleware/kafka/producer.html#重试机制" class="sidebar-link reco-side-重试机制" data-v-cb1513f6>重试机制</a></li></ul></main> <!----></div></div></div></div><div class="global-ui"><div class="back-to-ceiling" style="right:1rem;bottom:6rem;width:2.5rem;height:2.5rem;border-radius:.25rem;line-height:2.5rem;display:none;" data-v-c6073ba8 data-v-c6073ba8><svg t="1574745035067" viewBox="0 0 1024 1024" version="1.1" xmlns="http://www.w3.org/2000/svg" p-id="5404" class="icon" data-v-c6073ba8><path d="M526.60727968 10.90185116a27.675 27.675 0 0 0-29.21455937 0c-131.36607665 82.28402758-218.69155461 228.01873535-218.69155402 394.07834331a462.20625001 462.20625001 0 0 0 5.36959153 69.94390903c1.00431239 6.55289093-0.34802892 13.13561351-3.76865779 18.80351572-32.63518765 54.11355614-51.75690182 118.55860487-51.7569018 187.94566865a371.06718723 371.06718723 0 0 0 11.50484808 91.98906777c6.53300375 25.50556257 41.68394495 28.14064038 52.69160883 4.22606766 17.37162448-37.73630017 42.14135425-72.50938081 72.80769204-103.21549295 2.18761121 3.04276886 4.15646224 6.24463696 6.40373557 9.22774369a1871.4375 1871.4375 0 0 0 140.04691725 5.34970492 1866.36093723 1866.36093723 0 0 0 140.04691723-5.34970492c2.24727335-2.98310674 4.21612437-6.18497483 6.3937923-9.2178004 30.66633723 30.70611158 55.4360664 65.4791928 72.80769147 103.21549355 11.00766384 23.91457269 46.15860503 21.27949489 52.69160879-4.22606768a371.15156223 371.15156223 0 0 0 11.514792-91.99901164c0-69.36717486-19.13165746-133.82216804-51.75690182-187.92578088-3.42062944-5.66790279-4.76302748-12.26056868-3.76865837-18.80351632a462.20625001 462.20625001 0 0 0 5.36959269-69.943909c-0.00994388-166.08943902-87.32547796-311.81420293-218.6915546-394.09823051zM605.93803103 357.87693858a93.93749974 93.93749974 0 1 1-187.89594924 6.1e-7 93.93749974 93.93749974 0 0 1 187.89594924-6.1e-7z" p-id="5405" data-v-c6073ba8></path><path d="M429.50777625 765.63860547C429.50777625 803.39355007 466.44236686 1000.39046097 512.00932183 1000.39046097c45.56695499 0 82.4922232-197.00623328 82.5015456-234.7518555 0-37.75494459-36.9345906-68.35043303-82.4922232-68.34111062-45.57627738-0.00932239-82.52019037 30.59548842-82.51086798 68.34111062z" p-id="5406" data-v-c6073ba8></path></svg></div></div></div>
    <script src="/oliver-vuepress/assets/js/app.c21e6ffc.js" defer></script><script src="/oliver-vuepress/assets/js/3.6dd9a2a1.js" defer></script><script src="/oliver-vuepress/assets/js/1.898920d0.js" defer></script><script src="/oliver-vuepress/assets/js/44.10d7fe47.js" defer></script>
  </body>
</html>
